package com.l.im.service.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.l.im.common.api.ApiResponse;
import com.l.im.common.cache.RedisService;
import com.l.im.common.config.ConfigReader;
import com.l.im.common.thread.AsyncExecutor;
import com.l.im.model.SingleMessageDTO;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientException;

import java.util.List;

@Slf4j
public class MsgChatHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) throws Exception {
        Boolean isAuthenticated = (Boolean) ctx.channel().attr(AttributeKey.valueOf("isAuthenticated")).get();

        if (isAuthenticated == null || !isAuthenticated) {
            sendChannelResponse(ctx.channel(), ApiResponse.fail(401, "Connection not authenticated"));
            ctx.close();
            return;
        }

        JSONObject json = JSONObject.parseObject(frame.text());
        String type = json.getString("type");

        if ("login".equals(type)) {
            handleLogin(ctx, json);
        } else if ("single".equals(type)) {
            handleSingleMessage(ctx, json);
        }
    }

    private void handleLogin(ChannelHandlerContext ctx, JSONObject json) {
        String userId = json.getString("userId");
        String username = json.getString("username");
        String password = json.getString("password");

        if (userId != null && username != null && password != null) {
            MessageDispatcher.register(userId, ctx.channel());

            RedisService redisService = new RedisService();
            String topic = ConfigReader.getInstance().getOrDefault("rocketmq.single-consumer-topic", String.class, "");
            redisService.set("online:user:" + userId, userId + ":" + topic, 3600L);

            // 立即响应登录成功
            sendChannelResponse(ctx.channel(), ApiResponse.success("登录成功"));

            // 异步推送离线消息
            AsyncExecutor.execute(() -> pushOfflineMessages(userId));
        } else {
            sendChannelResponse(ctx.channel(), ApiResponse.fail(400, "登录信息不完整"));
        }
    }

    private void pushOfflineMessages(String userId) {
        List<String> offlineMessages = MessageDispatcher.getOfflineMessages(userId);
        if (offlineMessages != null && !offlineMessages.isEmpty()) {
            for (String msgJson : offlineMessages) {
                ApiResponse<?> response = ApiResponse.success(JSONObject.parse(msgJson));
                MessageDispatcher.sendToUser(userId, response);
            }
            log.info("向用户 {} 下发了 {} 条离线消息", userId, offlineMessages.size());
        }
    }


    private void handleSingleMessage(ChannelHandlerContext ctx, JSONObject json) throws ClientException {
        String fromUserId = json.getString("userId");
        String toUserId = json.getString("toUserId");
        String message = json.getString("message");

        MessageDispatcher.dispatchToUser(fromUserId, toUserId, message);
        sendChannelResponse(ctx.channel(), ApiResponse.success(new SingleMessageDTO("回复确认", fromUserId, toUserId, message)));
    }

    private void sendChannelResponse(Channel channel, ApiResponse<?> response) {
        channel.writeAndFlush(new TextWebSocketFrame(response.toString()));
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        log.info("连接建立: {}", ctx.channel().id());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        log.info("连接断开: {}", ctx.channel().id());
        MessageDispatcher.unregister(ctx.channel());
    }


}
